package mt.mahoutTest;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.mahout.clustering.WeightedVectorWritable;
import org.apache.mahout.clustering.kmeans.Cluster;
import org.apache.mahout.clustering.kmeans.KMeansDriver;
import org.apache.mahout.common.distance.EuclideanDistanceMeasure;
import org.apache.mahout.math.RandomAccessSparseVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;

public class KMeansClusterTest
{
    public static final double[][] points = {{1, 1}, {2, 1}, {1, 2}, {2, 2}, {3, 3}, {8, 8}, {9, 8}, {8, 9}, {9, 9}};
    
    public static void writePointsToFile(List<Vector> points, String fileName, FileSystem fs, Configuration conf) throws IOException {
    	Path path = new Path(fileName);
    	SequenceFile.Writer writer = new SequenceFile.Writer(fs, conf, path, LongWritable.class, VectorWritable.class);
    	long recNum = 0;
    	VectorWritable vec = new VectorWritable();
    	for (Vector point : points){
    		vec.set(point);
    		writer.append(new LongWritable(recNum++), vec);
    	}
    	writer.close();
    }
    
    public static List<Vector> getPoints(double[][] raw){
    	List<Vector> points = new ArrayList<Vector>();
    	for(int i=0; i<raw.length; i++){
    		double[] fr = raw[i];
    		Vector vec = new RandomAccessSparseVector(fr.length);
    		vec.assign(fr);
    		points.add(vec);
    	}
    	return points;
    }
    
    public static void main(String[] args) throws Exception{
    	int k = 2;
    	List<Vector> vectors = getPoints(points);
    	Path testData = new Path("testdata");
    	testData = new Path("testdata/points");

    	Configuration conf = new Configuration();
    	FileSystem fs = FileSystem.get(conf);
    	writePointsToFile(vectors, "testdata/points/file1", fs, conf);
    	
    	Path path =new Path("testdata/clusters/part-00000");
    	SequenceFile.Writer writer = new SequenceFile.Writer(fs, conf, path, Text.class, Text.class);
    	for(int i=0; i<k; i++){
    		Vector vec = vectors.get(i);
    		Cluster cluster = new Cluster(vec, i, new EuclideanDistanceMeasure());
    		writer.append(new Text(cluster.getIdentifier()), cluster);
    	}
    	writer.close();
    	
    	KMeansDriver.run(conf, new Path("testdata/points"), new Path("testdata/clusters"), new Path("output"), new EuclideanDistanceMeasure(), 0.001, 10, true, false);
    	
    	SequenceFile.Reader reader = new SequenceFile.Reader(fs, new Path("output/" + Cluster.CLUSTERED_POINTS_DIR + "/part-m-00000"), conf);
    	IntWritable key = new IntWritable();
    	WeightedVectorWritable value = new WeightedVectorWritable();
    	while(reader.next(key, value)){
    		System.out.println(value.toString() + " belongs to cluster " + key.toString());
    	}
    	reader.close();
    }
    
	/*
	private static String basePath = "/test";
    private static String inputDataPath = basePath+"/points";
    private static String outputDataPath = basePath+"/output";
    private static String baseClusterPath = basePath+"/clusters";
    private static int clusterCount = 2;
    private Configuration conf;
    private FileSystem fs;

    
    public KMeansClusterTest(){
        try {
            this.conf = getConfiguration();
            this.fs = FileSystem.get(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void writePointsToFile(List<NamedVector> points, String fileName) throws IOException {
        Path path = new Path(fileName);
        SequenceFile.Writer writer = new SequenceFile.Writer(this.fs, this.conf,  path,  Text.class, VectorWritable.class);
        VectorWritable vec = new VectorWritable();
        for(NamedVector point : points){
            vec.set(point);
            writer.append(new Text(point.getName()), vec);
        }
        writer.close();
    }

    private void writeInitialCenterPoints(List<NamedVector> points) throws IOException {
        
        Path path = new Path(baseClusterPath+"/part-00000");
        SequenceFile.Writer writer = new SequenceFile.Writer(this.fs,  this.conf, path, Text.class, Cluster.class);
        
        for(int i=0;i<clusterCount;i++){
            Vector vec = points.get(i);
            Cluster cluster = new Cluster(vec, i, new EuclideanDistanceMeasure());
            writer.append(new Text(cluster.getIdentifier()), cluster);
        }
        writer.close();
    }
    
    private List<NamedVector> getPoints(){
        List<NamedVector> points = new ArrayList<NamedVector>();
        points.add(new NamedVector(new DenseVector(new double[]{1,1}), "학교"));
        points.add(new NamedVector(new DenseVector(new double[]{2,1}), "종이"));
        points.add(new NamedVector(new DenseVector(new double[]{3,2}), "땡땡땡"));
        points.add(new NamedVector(new DenseVector(new double[]{2,2}), "어서"));
        points.add(new NamedVector(new DenseVector(new double[]{3,3}), "모이자"));
        points.add(new NamedVector(new DenseVector(new double[]{8,2}), "선생님이"));
        points.add(new NamedVector(new DenseVector(new double[]{9,8}), "우리를"));
        points.add(new NamedVector(new DenseVector(new double[]{8,9}), "기다리"));
        points.add(new NamedVector(new DenseVector(new double[]{9,9}), "신다"));
        return points;
    }
    
    private Configuration getConfiguration() throws IOException{
        Configuration conf = new Configuration();
        Resource resource1 = new ClassPathResource("conf/core-site.xml");
        Resource resource2 = new ClassPathResource("conf/mapred-site.xml");
        Resource resource3 = new ClassPathResource("conf/hdfs-site.xml");
        conf.addResource(resource1.getURL());
        conf.addResource(resource2.getURL());
        conf.addResource(resource3.getURL());
        
        return conf;
    }
    
    public void execute() throws Exception{
        
        //데이터 세팅
        List<NamedVector> vectors = getPoints();
        
        // sequence file 생성
        writePointsToFile(vectors, inputDataPath+"/file1"); 
        
        // 초기 기준 점 찍기
        writeInitialCenterPoints(vectors);  
        
        // 실행 
        KMeansDriver.run(conf,  new Path(inputDataPath), new Path(baseClusterPath), new Path(outputDataPath),
            new EuclideanDistanceMeasure(), 0.001,10,true, false);
    }

    public void confirmResult() throws Exception{
        SequenceFile.Reader reader = new SequenceFile.Reader(fs,  new Path(outputDataPath+"/"+Cluster.CLUSTERED_POINTS_DIR+"/part-m-00000"), conf);
        IntWritable key = new IntWritable();
        WeightedVectorWritable value = new WeightedVectorWritable();
        while(reader.next(key, value)){
            NamedVector v = (NamedVector)value.getVector();
            System.out.println(key.toString()+ " : "+v.getName());
            System.out.println(key.toString()+ " : "+value.toString());
        }
        reader.close();
    }

    public static void main(String[] args) throws Exception{
        KMeansClusterTest test = new KMeansClusterTest();
        test.execute();
        test.confirmResult();
    }
    */
}
